package com.dgw.springbootkafka.receiver;

import org.apache.kafka.clients.admin.Admin;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaAdmin;

import java.util.ArrayList;
import java.util.List;

/**
 * @program: springboot-kafka
 * @description:
 * @author: Mr.Dai
 * @create: 2020-02-22 23:20
 **/
public class CommitConsumer {

    @KafkaListener(topics = "test-syn", groupId = "test-2")
    public void send3(KafkaConsumer consumer) {
        List<String> topics = new ArrayList<>();
        topics.add("test-syn");
        consumer.subscribe(topics);
        for (; ; ) {
            // 拉取任务超时时间
            ConsumerRecords<String, String> records = consumer.poll(1000);
            for (ConsumerRecord consumerRecord : records) {
                System.out.println("partition:" + consumerRecord.partition());
                System.out.println("offset:" + consumerRecord.offset());
                System.out.println("key:" + consumerRecord.key());
                System.out.println("value:" + consumerRecord.value());
            }
            // 当前批次offset
            consumer.commitSync();
        }
    }
}
